package com.arnold.test.demos.fileStream

import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.{SparkConf, SparkContext}

/**
  * Created by arnold.zhu on 2017/7/18.
  *
  * <p>Spark Streaming将监控某目录中的文件，<b>获取在间隔时间段内变化的数据</b>，然后通过Spark Streaming计算出改时间段内单词统计数。</p>
  */
object FileWordCount {

  def main(args: Array[String]): Unit = {
    // 创建Streaming的上下文，包括Spark的配置和时间间隔，这里时间为间隔5秒
    val sc = new StreamingContext(new SparkConf(), Seconds(3))
    // 指定监控的目录，在这里为/Users/arnold.zhu/files
    val lines = sc.textFileStream("/opt/wordcount/")
    // 对指定文件夹变化的数据进行单词统计并且打印
    val words = lines.flatMap(_.split(""))
    // 转换成 一个个的map(Key,1)形式 reduceByKey返回一个（K，V）对的数据集
    val wordCounts = words.map(x => (x, 1)).reduceByKey((x, y) => x + y)
    // 将输出时间戳的同时将输出单词统计个数。
    wordCounts.print()
    // 启动Streaming
    sc.start()
    sc.awaitTermination()
  }


}
